package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"labrpc"
	"math/rand"
	"sync"
	"time"
)

// import "bytes"
// import "encoding/gob"

const (
	FOLLOWER = iota
	CANDIDATE
	LEADER
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make().
//
type ApplyMsg struct {
	Index       int
	Command     interface{}
	UseSnapshot bool   // ignore for lab2; only used in lab3
	Snapshot    []byte // ignore for lab2; only used in lab3
}
type LogEntry struct {
	Index int
	Term  int
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex
	peers     []*labrpc.ClientEnd
	persister *Persister
	me        int // index into peers[]
	state     int
	// Your data here.
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	currentTerm int
	votedFor    int
	log         []LogEntry

	// Volatile state on all server
	commitIndex int
	lastApplied int

	// Volatile state on leaders
	nextIndex  []int
	matchIndex []int

	electionTimeout   *time.Timer
	appendEntriesChan chan bool
	requestVoteChan   chan bool
	stop              bool
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	// Your code here.
	return rf.currentTerm, rf.state == LEADER
}

func (rf *Raft) getLastLogIndex() int {
	return rf.log[len(rf.log)-1].Index
}
func (rf *Raft) getLastLogTerm() int {
	return rf.log[len(rf.log)-1].Term
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here.
	// Example:
	// w := new(bytes.Buffer)
	// e := gob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	// Your code here.
	// Example:
	// r := bytes.NewBuffer(data)
	// d := gob.NewDecoder(r)
	// d.Decode(&rf.xxx)
	// d.Decode(&rf.yyy)
}

//
// example RequestVote RPC arguments structure.
//
type RequestVoteArgs struct {
	// Your data here.
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
//
type RequestVoteReply struct {
	// Your data here.
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here.
	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.requestVoteChan <- true
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		return
	}
	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term
		reply.Term = rf.currentTerm
		rf.votedFor = -1
		rf.state = FOLLOWER
	}

	if args.Term == rf.currentTerm {
		index := rf.getLastLogIndex()
		term := rf.getLastLogTerm()
		if (rf.votedFor == args.CandidateId || rf.votedFor == -1) && index == args.LastLogIndex && term == args.LastLogTerm {
			reply.VoteGranted = true
			rf.votedFor = args.CandidateId
		}
	}
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// returns true if labrpc says the RPC was delivered.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	prevLogIndex int
	prevLogTerm  int
	Entries      []LogEntry
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term      int
	Success   bool
	NextIndex int
}

func (rf *Raft) AppendEntries(args AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	DPrintf("%v:appendEntries start\n", rf.me)
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		reply.Success = false
		return
	}
	rf.appendEntriesChan <- true
	rf.currentTerm = args.Term
	rf.state = FOLLOWER
	rf.votedFor = -1

	if args.Term > rf.currentTerm {
		reply.Term = rf.currentTerm
		reply.Success = false
		return
	}
	if args.prevLogIndex > rf.getLastLogIndex() {
		reply.Success = false
		reply.NextIndex = rf.getLastLogIndex() + 1
		return
	}
	ok := false
	index := 0
	for i, info := range rf.log {
		if info.Index == args.prevLogIndex && info.Term == args.Term {
			ok = true
			reply.NextIndex = info.Index + 1
			index = i
			break
		}
	}
	if ok {
		reply.Success = true
		rf.log = append(rf.log[:index], args.Entries...)
		return
	}
	if args.LeaderCommit > rf.commitIndex {
		if args.LeaderCommit > rf.getLastLogIndex() {
			rf.commitIndex = rf.getLastLogIndex()
		} else {
			rf.commitIndex = args.LeaderCommit
		}
	}
}

func (rf *Raft) sendAppendEntries(server int, args AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	return index, term, isLeader
}

//
// the tester calls Kill() when a Raft instance won't
// be needed again. you are not required to do anything
// in Kill(), but it might be convenient to (for example)
// turn off debug output from this instance.
//
func (rf *Raft) Kill() {
	// Your code here, if desired.
	rf.stop = true
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
// 建一个Raft端点。
// peers参数是通往其他Raft端点处于连接状态下的RPC连接。
// me参数是自己在端点数组中的索引。
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.electionTimeout = time.NewTimer(rf.randomTime())
	rf.state = FOLLOWER
	rf.log = append(rf.log, LogEntry{
		Term:  1,
		Index: 1,
	})
	rf.appendEntriesChan = make(chan bool, 10)
	rf.requestVoteChan = make(chan bool, 10)
	rf.resetTimeout()
	// Your initialization code here.

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	go rf.loop()

	return rf
}

func (rf *Raft) loop() {
	for !rf.stop {
		switch rf.state {
		case FOLLOWER:
			select {
			case <-rf.appendEntriesChan:
				rf.resetTimeout()
			case <-rf.requestVoteChan:
				rf.resetTimeout()
			case <-time.After(time.Duration(rand.Int63n(500)+500) * time.Millisecond):
				DPrintf("%v:FOLLOWER time out!\n", rf.me)
				rf.state = CANDIDATE
			}
		case CANDIDATE:
			DPrintf("%v:CANDIDATE start\n", rf.me)
			rf.currentTerm++
			rf.votedFor = rf.me
			leaderChan := make(chan bool, 1)
			go rf.broadcastRequestVote(leaderChan)
			select {
			case ok := <-leaderChan:
				if ok {
					rf.state = LEADER
				}
			case <-rf.appendEntriesChan:
				rf.state = FOLLOWER
				rf.votedFor = -1
			case <-time.After(time.Duration(rand.Int63n(500)+500) * time.Millisecond):
				DPrintf("%v:CANDIDATE time out\n", rf.me)
			}
		case LEADER:
			DPrintf("LEADER:%v-%v", rf.me, rf.currentTerm)
			rf.broadcastAppendEntries()
			select {
			case <-time.After(100 * time.Millisecond):
			}
		}
	}
}

func (rf *Raft) resetTimeout() {
	rf.electionTimeout.Reset(rf.randomTime())
}

func (rf *Raft) randomTime() time.Duration {
	return time.Duration(rand.Intn(300))*time.Millisecond + 500*time.Millisecond
}

func (rf *Raft) broadcastRequestVote(leaderChan chan bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	voteCount := 1
	for i := range rf.peers {
		if i != rf.me && rf.state != LEADER {
			args := RequestVoteArgs{
				Term:         rf.currentTerm,
				CandidateId:  rf.me,
				LastLogIndex: rf.getLastLogIndex(),
				LastLogTerm:  rf.getLastLogTerm(),
			}
			reply := &RequestVoteReply{}
			if ok := rf.sendRequestVote(i, args, reply); ok {
				DPrintf("%v-%v:reply %v", rf.me, rf.currentTerm, reply)
				if reply.Term > rf.currentTerm {
					rf.currentTerm = reply.Term
					rf.state = FOLLOWER
				}
				if reply.Term == rf.currentTerm && reply.VoteGranted {
					voteCount++
					DPrintf("xxxx:%v,%v\n", voteCount, len(rf.peers)/2)
					if voteCount > len(rf.peers)/2 {
						leaderChan <- true
						return
					}
				}
			}
		}
	}
	// leaderChan <- false
}
func (rf *Raft) broadcastAppendEntries() {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	successCount := 0
	for i := range rf.peers {
		if i != rf.me && rf.state == LEADER {
			args := AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderId:     rf.me,
				prevLogIndex: rf.getLastLogIndex(),
				prevLogTerm:  rf.getLastLogTerm(),
			}
			reply := &AppendEntriesReply{}
			if ok := rf.sendAppendEntries(i, args, reply); ok {
				if reply.Term > rf.currentTerm {
					rf.state = FOLLOWER
					rf.votedFor = -1
					return
				}
				if reply.Term < rf.currentTerm {
					continue
				}
				if reply.Success {
					successCount++
				}
			}
		}
	}
	if successCount > len(rf.peers)/2 {
		//　copy log finished. commit
	}
}
